package org.example.message_queue.mqclient;

import ch.qos.logback.classic.spi.EventArgUtil;
import org.example.message_queue.common.*;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Connection {
//    要进行网络通信
    private Socket socket = null;
//    管理多个channel
    ConcurrentHashMap<String,Channel> channelMap = new ConcurrentHashMap<>();
//    读写操作
    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;

    private ExecutorService callbackPool = null;

    public Connection(String host, int port) throws IOException {
        socket = new Socket(host,port);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);
        callbackPool = Executors.newFixedThreadPool(4);

//        创建扫描线程，由这个扫描线程不断从socket中读取响应数据，将响应数据再交给对应的channel处理
        Thread thread = new Thread(()->{
            try{
                while (!socket.isClosed()){

//          读取响应数据
//                    有两种返回情况：一种是客户端给服务器发送请求，服务器完成后返回结果给客户端；另外一种是服务器有消息，给客户端推送消息
                    Response response = readResponse();
                    dispatchResponse(response);
                }
            } catch (SocketException e){
//            连接正常关闭，可以忽略这个异常
                System.out.println("[Connection] 连接正常断开！");
            } catch (IOException | ClassNotFoundException | MqException e) {
                e.printStackTrace();
                System.out.println("[Connection] 连接异常断开！");
            }
        });
        thread.start();
    }

//    关闭connection，释放资源
    public void close(){
        try{
            channelMap.clear();
            callbackPool.shutdownNow();
            socket.close();
            inputStream.close();
            outputStream.close();
        }catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
//        如何进行区分：type
        if (response.getType() == 0xc){
            System.out.println("不进入这里");
//        服务器推送的消息数据
//            运行这一步报错是吗？ 不是
            SubScribeReturns subScribeReturns = (SubScribeReturns)BinaryTool.fromBytes(response.getPayload());
//            根据channelId找到对应的channel对象
            Channel channel = channelMap.get(subScribeReturns.getChannelId());
            if (channel == null){
                throw new MqException("[Connection] 该消息对应的channel在客户端中不存在！ channelId="+subScribeReturns.getChannelId());
            }
//            如果直接在扫描线程中执行这个回调函数,时间上会有点长，可能造成阻塞，所以这里可以交给线程池去处理
//            channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(), subScribeReturns.getBasicProperties(), subScribeReturns.getBody());
            callbackPool.submit(() ->{
                try {
                    channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(), subScribeReturns.getBasicProperties(), subScribeReturns.getBody());
                } catch (MqException | IOException e) {
                    e.printStackTrace();
                }
            });
        }else {
//        客户端给服务器发送请求，服务器给客户端响应结果
//            先解析响应，再将响应写入对应channel的basicReturnsMap中。后续客户端找响应就会在这个hashmap中找对应的响应
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());

            Channel channel = channelMap.get(basicReturns.getChannelId());

            if (channel == null){
                throw new MqException("[Connection]该消息对应的channel不在客户端中！ channelId="+basicReturns.getChannelId());
            }
            channel.putReturns(basicReturns);
        }
    }

    //    发送请求
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
        System.out.println("[Connection] 发送请求！type=" + request.getType() + ",length=" + request.getLength() + ",payload=" + request.getPayload());
    }

//    读取响应
    public Response readResponse() throws IOException {
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] payload = new byte[response.getLength()];
        int n = dataInputStream.read(payload);
        if (n != response.getLength()){
            throw new IOException("读取的响应数据不完整！");
        }
        response.setPayload(payload);
        System.out.println("[Connection]收到响应! type=" + response.getType() + ",length=" + response.getLength() + ",payload=" + response.getPayload());
        return response;
    }

//    创建channel
    public Channel createChannel() throws IOException {
        String channelId = "C-"+ UUID.randomUUID().toString();
        Channel channel = new Channel(channelId,this);
        channelMap.put(channelId,channel);
        boolean ok = channel.createChannel();
        if (!ok){
            channelMap.remove(channelId,channel);
            return null;
        }
        return channel;
    }
}
